package com.codejiwei.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * author: codejiwei
 * date: 2023/8/11
 * desc: flink sql with grouping sets
 **/
public class Flink_SQL_GroupingSets {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

//        String sql = "SELECT supplier_id, rating, COUNT(*) AS total\n" +
//                "FROM (VALUES\n" +
//                "    ('supplier1', 'product1', 4),\n" +
//                "    ('supplier1', 'product2', 3),\n" +
//                "    ('supplier2', 'product3', 3),\n" +
//                "    ('supplier2', 'product4', 4))\n" +
//                "AS Products(supplier_id, product_id, rating)\n" +
//                "GROUP BY GROUPING SETS ((supplier_id, rating), (supplier_id), ())";


        String sql = "SELECT supplier_id, product_id, SUM(rating)\n" +
                "FROM (VALUES\n" +
                "    ('supplier1', 'product1', 4),\n" +
                "    ('supplier1', 'product2', 3),\n" +
                "    ('supplier2', 'product3', 3),\n" +
                "    ('supplier2', 'product4', 4))\n" +
                "AS Products(supplier_id, product_id, rating)\n" +
                "GROUP BY ROLLUP (supplier_id, product_id)";

//        String sql = "SELECT supplier_id, rating, COUNT(*)\n" +
//                "FROM (VALUES\n" +
//                "    ('supplier1', 'product1', 4),\n" +
//                "    ('supplier1', 'product2', 3),\n" +
//                "    ('supplier2', 'product3', 3),\n" +
//                "    ('supplier2', 'product4', 4))\n" +
//                "AS Products(supplier_id, product_id, rating)\n" +
//                "GROUP BY CUBE (supplier_id, product_id)";


        Table table = tableEnv.sqlQuery(sql);
        tableEnv.toChangelogStream(table).print();

        env.execute();

    }
}
